package com.zxymq.amqp.listener;

import com.fasterxml.jackson.core.type.TypeReference;
import com.rabbitmq.client.Channel;
import com.zxymq.amqp.constants.MQConstants;
import com.zxymq.amqp.support.DefaultJackson2JsonMessageConverter;
import com.zxymq.amqp.support.ZxyJackson2JavaTypeMapper;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.Assert;

import java.util.*;

/**
 * RabbitMQ抽象消息监听，所有消息消费者必须继承此类
 *
 * @author:
 * @date :2018/1/11
 */
public abstract class AbstractMessageListener implements ChannelAwareMessageListener {
    //检测是否复杂类型，如果是，则开启TypeReference
    private boolean open;
    private TypeReference typeReference;
    //定义基本数据类型策略
    private Map<String, Class<?>> baseClassMap = new HashMap<String, Class<?>>() {
        {
            put(Boolean.class.toString(), Boolean.class);
            put(Character.class.toString(), Character.class);
            put(Byte.class.toString(), Byte.class);
            put(Short.class.toString(), Short.class);
            put(Integer.class.toString(), Integer.class);
            put(Long.class.toString(), Long.class);
            put(Float.class.toString(), Float.class);
            put(Double.class.toString(), Double.class);
            put(String.class.toString(), String.class);
          /*  put(Map.class.toString(), Map.class);
            put(List.class.toString(), List.class);*/
        }
    };

    private void setComplexGeneric(boolean open, TypeReference typeReference) {
        this.open = open;
        this.typeReference = typeReference;
    }

    /**
     * 功能描述:设置复杂泛型处理
     *
     * @param: [typeReference]
     * @return: void
     * @auther: wfm0105@163.com
     * @date: 2018/8/9 15:58
     */
    public void setComplexGeneric(TypeReference typeReference) {
        Assert.notNull(typeReference, "'targetTypeReference' must not be null");
        this.setComplexGeneric(true, typeReference);
    }


    protected Logger logger = LoggerFactory.getLogger(this.getClass());
    /**
     * 消费次数
     */
    private Integer maxConsumerCount = 3;

    @Autowired
    private DefaultJackson2JsonMessageConverter messageConverter;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    /**
     * 目标转换类(默认为Object)
     */
    @Setter
    private Class<?> dstClass = Object.class;


    /**
     * 接收消息，子类必须实现该方法
     *
     * @param message          消息对象
     * @param messageConverter 消息转换器（对象直接会转换成想要的对象信息）
     */
    public abstract void receiveMessage(Message message, MessageConverter messageConverter);

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        MessageProperties messageProperties = message.getMessageProperties();
        Long deliveryTag = messageProperties.getDeliveryTag();
        Long consumerCount = redisTemplate.opsForHash().increment(MQConstants.MQ_CONSUMER_RETRY_COUNT_KEY,
                messageProperties.getMessageId(), 1);

        logger.info("当前消息ID:{} 消费次数：{}", messageProperties.getMessageId(), consumerCount);

        try {
            if (messageProperties.getHeaders().get(ZxyJackson2JavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME) != null && (dstClass == null)) {
                throw new RuntimeException("Received message does not have a target conversion class configured");
            }
            //基本数据了类型转换映射
            boolean isbaseClass = messageProperties.getHeaders().get("baseClass").toString().contains("java.lang");
            if (isbaseClass) {
                dstClass = getDstClass(messageProperties.getHeaders().get("baseClass").toString());
            }
            //消费端配置映射（简单的泛型可以直接通过此方式处理）
            if (dstClass != null) {
                ZxyJackson2JavaTypeMapper jackson2JavaTypeMapper = new ZxyJackson2JavaTypeMapper();
                jackson2JavaTypeMapper.setIdClassMapping(dstClass);
                messageConverter.setJavaTypeMapper(jackson2JavaTypeMapper);
            } else if (null != this.typeReference) { //消费端配置映射（简单和复杂的方式都可以通过此方式处理）
                messageConverter.setTypePrecedence(this.typeReference);
            }
            receiveMessage(message, messageConverter);
            // 成功的回执
            channel.basicAck(deliveryTag, false);
            // 如果消费成功，将Redis中统计消息消费次数的缓存删除
            redisTemplate.opsForHash().delete(MQConstants.MQ_CONSUMER_RETRY_COUNT_KEY,
                    messageProperties.getMessageId());
        } catch (Exception e) {
            logger.error("RabbitMQ 消息消费失败，" + e.getMessage(), e);
            if (consumerCount >= maxConsumerCount) {
                // 入死信队列
                channel.basicReject(deliveryTag, false);
            } else {
                // 重回到队列，重新消费
                channel.basicNack(deliveryTag, false, true);
            }
        }
    }

    public Integer getMaxConsumerCount() {
        return maxConsumerCount;
    }

    public void setMaxConsumerCount(Integer maxConsumerCount) {
        this.maxConsumerCount = maxConsumerCount;
    }

    private Class<?> getDstClass(String classStr) {
        Class dstClass = baseClassMap.get(classStr);
        return dstClass;
    }
}
